package com.duowan.realtime.model.builder;

import com.duowan.realtime.model.SqlJob;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @version 1.0
 * @author: tangsicheng
 * @date: 14-8-28
 * @since 1.0
 */
public class SqlJobBuilderImpl implements SqlJobBuilder {

    private SqlJob sqlJob = new SqlJob();

    private static Map<String,String> KPICODE_AGGREFUNC_MAP = new HashMap<String, String>(){{
        put("mid_new_user_ucnt",    "hll_count_distinct(mid,'history')");      //新增机器
        put("new_user_ucnt",        "hll_count_distinct(passport,'history')"); //新增用户
        put("mid_active_user_ucnt", "count_distinct(mid,'day')");              //活跃机器
        put("active_user_ucnt",     "count_distinct(passport,'day')");         //活跃机器
        put("startup_time",         "count(startupEvent)");                               //启动次数
    }};

    private static Map<String,String> KPICODE_WHERE_CONDITION = new HashMap<String, String>(){{
        put("startup_time", " where startupEvent ='true' ");
    }};

    private static Map<String,String> KPICODE_GROUPBY_COLLUMN = new HashMap<String, String>(){{
        put("mid_active_user_ucnt", "format(stime,'yyyyMMdd') timeStamp,");              //活跃机器
        put("active_user_ucnt",     "format(stime,'yyyyMMdd') timeStamp,");         //活跃机器
    }};

    private static String SELECT = "select max(stime) timeStamp, ";

    private static String FROM_TABLE = " from ActionLogTable ";

    private static String GROUP_BY = "group by ";

    private static String REDIS_KEY = "var key = RedisKey(format(row.timeStamp,'yyyyMMdd'), ";

    private static String REDIS_TIMESTAMP_KEY = "var keyTimeStamp = key+'_timeStamp';";

    private static String REDIS_INCR = "redis.incrBy(key,(long)row.$kpiCode$);";

    private static String REDIS_SET = "redis.set(keyTimeStamp,format(row.timeStamp,'yyyyMMdd HH:mm:ss'));";



    @Override
    public void buildSqlName(String sqlName) {
        sqlJob.setName(sqlName);

    }

    @Override
    public void buildSQLStatement(String kpiCode, List<String> dimList) {
        StringBuilder sb = new StringBuilder();
        sb.append(SELECT);
        sb.append(KPICODE_GROUPBY_COLLUMN.get(kpiCode) !=null ?KPICODE_GROUPBY_COLLUMN.get(kpiCode):"" );
        for(String dim:dimList){
            sb.append(dim + ", ");
        }
        sb.append(KPICODE_AGGREFUNC_MAP.get(kpiCode)+" as "+ kpiCode);
        sb.append(FROM_TABLE);
        sb.append(KPICODE_WHERE_CONDITION.get(kpiCode) !=null ?KPICODE_WHERE_CONDITION.get(kpiCode):"");
        sb.append(GROUP_BY);
        sb.append(KPICODE_GROUPBY_COLLUMN.get(kpiCode) !=null ?KPICODE_GROUPBY_COLLUMN.get(kpiCode):"" );
        for(String dim:dimList){
            sb.append( dim + ", ");
        }
        sqlJob.setQueryStatement(sb.substring(0,sb.lastIndexOf(",")));
    }


    @Override
    public void buildIntoRedis(String kpiCode, List<String> dimList) {
        String redisKeyStr = getRedisKey(kpiCode, dimList);
        String redisIncrByExpress = getRedisIncrByExpress(kpiCode);
        sqlJob.setIntoRedis(redisKeyStr+redisIncrByExpress+REDIS_TIMESTAMP_KEY+REDIS_SET);
    }

    private String getRedisIncrByExpress( String kpiCode) {
        return REDIS_INCR.replace("$kpiCode$",kpiCode);
    }

    private String getRedisKey(String kpiCode, List<String> dimList) {
        StringBuilder sb = new StringBuilder();
        sb.append(REDIS_KEY);
        sb.append(addSingleQuote(kpiCode)+", ");


        //redis key目前的规则是 yyyymmdd|kpicode|维度1/维度1的值/维度2/维度2的值
        // 但是如果维度只是有product的话，必须也得加上iver，且值为null。
        // 维度只有game的话，必须也得加上gameServer，且值为null
        if(dimList.contains("game") && !dimList.contains("gameServer")){
            sb.append(addSingleQuote("game")+", ");
            sb.append("row.game, ");
            sb.append(addSingleQuote("gameServer")+", ");
            sb.append(addSingleQuote("null")+", ");
        }else if(dimList.contains("product") && !dimList.contains("iver")){
            sb.append(addSingleQuote("product")+", ");
            sb.append("row.product, ");
            sb.append(addSingleQuote("iver")+", ");
            sb.append(addSingleQuote("null")+", ");
        }
        else{
            for(String dim: dimList){
                sb.append(addSingleQuote(dim)+", ");
                sb.append("row."+dim+", ");
            }
        }
        StringBuilder sb2 = new StringBuilder(sb.substring(0,sb.lastIndexOf(",")));
        sb2.append(");");
        return sb2.toString();
    }

    private String addSingleQuote(String key){
        return  "'"+key+"'";
    }

    @Override
    public void buildIntoDb() {

    }

    @Override
    public SqlJob getResult() {
        return sqlJob;
    }
}
